
#include "common.h"
#include "thread.h"
#include "globals.h"
#include "http.h"
#include "server.h"

static int init_count = 0;
static pthread_mutex_t init_lock;
static pthread_cond_t init_cond;


static void *thread_loop(void *arg);
static void thread_event_setup(thread_t *t);
static void worker_thread_create(void* (*handler)(void *), void *arg);
static void thread_libevent_handle(int fd, short flag, void *arg);


void thread_nconn_add(thread_t *th)
{
	pthread_mutex_lock(&th->lock);
	th->nconn++;
	pthread_mutex_unlock(&th->lock);
}

void thread_nconn_del(thread_t *th)
{
	pthread_mutex_lock(&th->lock);
	th->nconn--;
	pthread_mutex_unlock(&th->lock);
}

thread_t *thread_select(void)
{
	int min ;
	int pos ;
	int i;
	
	min = threads[1].nconn;
	pos = 1;
	for(i = 1; i <= nworker; i++) {
		if(threads[i].nconn < min) {
			min = threads[i].nconn;
			pos = i;	
		}
	}
	
	thread_nconn_add(&threads[pos]);
	
	debug_log(3, "select thread %d, id %u\n", pos, threads[pos].id);
	return &threads[pos];
}

void thread_init(int number)
{
	int factual = 2;
	int i;
	
	pthread_mutex_init(&init_lock, NULL);
	pthread_cond_init(&init_cond, NULL);
	
	if(number > 0) 
		factual = number;
	
	nworker = factual - 1;

	threads = mempool_alloc(factual * sizeof(*threads));
	if(!threads) {
			error_log("init threads %d failed\n", factual);
			abort();
	}

	threads[0].base = master_base;
	threads[0].id = pthread_self();


	for (i = 0; i < factual; i++) {
		int fds[2];
		if (pipe(fds)) {
			error_log("Can't create notify pipe. %d\n", i);
			exit(1);
		}

		threads[i].notify_receive_fd = fds[0];
		threads[i].notify_send_fd = fds[1];
		threads[i].nconn = 0;
		pthread_mutex_init( &threads[i].lock, NULL);

		thread_event_setup(&threads[i]);
	}

	/* create thread */
	for(i = 1; i < factual; i++)
		worker_thread_create(thread_loop, &threads[i]);

	pthread_mutex_lock(&init_lock);
	init_count++; /* main thread */
	while (init_count < factual) {
		pthread_cond_wait(&init_cond, &init_lock);
	}
	pthread_mutex_unlock(&init_lock);

	debug_log(1, "1 master, %d worker running ...\n", nworker);
}

static void thread_event_setup(thread_t *me)
{
	if(!me) 
		return;

	if (! me->base) {
		me->base = event_init();
		if (! me->base) {
			fprintf(stderr, "Can't allocate event base\n");
			exit(1);
		}
	}

	/* Listen for notifications from other threads */
	event_set(&me->notify_event, me->notify_receive_fd,
						EV_READ | EV_PERSIST, thread_libevent_handle, me);
	event_base_set(me->base, &me->notify_event);

	if (event_add(&me->notify_event, 0) == -1) {
		fprintf(stderr, "Can't monitor libevent notify pipe\n");
			exit(1);
	}
	
	/* init conns queue */
	me->conns = (queue_t *)queue_create();
}

static void worker_thread_create(void * (*handler)(void *), void *arg)
{
	pthread_t       thread;
	pthread_attr_t  attr;
	int             ret;

	pthread_attr_init(&attr);

	if ((ret = pthread_create(&thread, &attr, handler, arg)) != 0) {
			fprintf(stderr, "Can't create thread: %s\n",
								strerror(ret));
			exit(1);
	}
}

static void * thread_loop(void *arg)
{
	assert(arg);
	
	thread_t *thread = arg;
	thread->id = pthread_self();

	pthread_mutex_lock(&init_lock);
	init_count++;
	pthread_cond_signal(&init_cond);
	pthread_mutex_unlock(&init_lock);

	event_base_loop(thread->base, 0);
	
	return;
}

static void thread_libevent_handle(int fd, short flag, void *arg)
{
	/* Create new connection. Add callback function to event. */
	assert(arg);

	queue_t *queue;
	char buf[1];
	thread_t *thread = arg;
	accept_t *at;
	struct list_head head;

	if (read(fd, buf, 1) != 1)
		debug_log(1, "fd %d Can't read from libevent pipe\n", fd);

	debug_log(3, "get notify from %u\n", threads[0].id);
	
	queue = thread->conns;
	pthread_mutex_lock(&queue->lock);
	while(list_empty(&queue->list)) 
		pthread_cond_wait(&queue->cond, &queue->lock);
	list_add(&head, &queue->list);
	list_del_init(&queue->list);
	queue->count = 0;
	pthread_mutex_unlock(&queue->lock);
	
	struct list_head *p;
	for(p = head.next; p!=&head; p = p->next) {
		conn_t *con;
		queue_item_t *entry;
		accept_t *at;

	 	entry = list_entry(p, queue_item_t, list);
		at = (accept_t *)entry->node;
		debug_log(3, "fd %d, flag %d, data %p, handler %p find\n", \
						at->fd, at->flag, at->data, at->handler);
		con = conn_new(at->fd, at->flag, at->handler, NULL, (struct in_addr *)&at->data, thread->base, thread);
		debug_log(3, "new conn %p\n", con);

		/* free some unused data */
		//mempool_free(at);
		//queue_item_free(entry);
	}
}
